package test

import com.gitee.ixtf.guice.Jguice
import com.gitee.ixtf.vertx.guice.VertxGuice
import com.gitee.ixtf.vertx.verticle.ConsumerCoroutineVerticle
import com.google.common.collect.Queues
import io.vertx.core.Vertx
import io.vertx.core.eventbus.EventBus
import io.vertx.core.eventbus.Message
import io.vertx.kotlin.core.vertxOptionsOf
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.tracing.opentelemetry.OpenTelemetryOptions
import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ObsoleteCoroutinesApi
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.TickerMode
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.channels.ticker
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext

private var scList = listOf<SendChannel<String>>()

private val vertx: Vertx by Jguice
private val eventBus: EventBus by Jguice

@OptIn(ObsoleteCoroutinesApi::class)
suspend fun main() {
  Jguice.init(VertxGuice(Vertx.vertx()))

  val queue = Queues.newPriorityBlockingQueue<String>()
  (1..10).forEach { queue.add("$it") }
  queue
  channelFlow { queue.forEach { launch { send(it) } } }.collect()
  println(queue.size)

  Vertx.vertx(
      vertxOptionsOf(
          preferNativeTransport = true,
          tracingOptions = OpenTelemetryOptions(),
      ))

  val alg1 = QueueVerticle()
  vertx.deployVerticle(alg1).coAwait()

  ticker(5.seconds.inWholeMilliseconds, mode = TickerMode.FIXED_DELAY).consumeEach {
    println(eventBus.request<Boolean>("test", "$it").coAwait().body())
  }

  println("success")
}

private class QueueVerticle : ConsumerCoroutineVerticle<String, Boolean>("test") {
  private val queue = Queues.newPriorityBlockingQueue<String>()

  override suspend fun start() {
    super.start()
    launch {
      while (true) {
        val poll = withContext(Dispatchers.IO) { queue.take() }
        log.info("handle $poll")
      }
    }
  }

  override suspend fun handle(reply: Message<String>): Boolean = queue.add(reply.body())
}
